package com.codejiwei.flink.table;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * author: codejiwei
 * date: 2023/8/8
 * desc: flink table api set run mode
 **/
public class Flink_Table_API_RunMode {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //set run mode
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //TODO table api only support STREAMING or BATCH
//        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStreamSource<Row> sourceDS = env.fromElements(
                Row.of("Tom", 22),
                Row.of("Jerry", 33),
                Row.of("Tom", 11)
        );

        Table table = tEnv.fromDataStream(sourceDS).as("name", "score");
        tEnv.createTemporaryView("SourceTable", table);

        Table sqlQuery = tEnv.sqlQuery("select name, sum(score) from SourceTable group by name");

        //TODO if run mode is STEAMING not support toDataStream
        DataStream<Row> rowDataStream = tEnv.toDataStream(sqlQuery);

//        DataStream<Row> rowDataStream = tEnv.toChangelogStream(sqlQuery);

        rowDataStream.print();
        env.execute();


    }
}
